package com.wksc.framework.impexp.thread;

import com.wksc.framework.impexp.bean.ImportExport;
import com.wksc.framework.impexp.constants.IEConstants;
import com.wksc.framework.impexp.constants.TaskStat;
import com.wksc.framework.impexp.log.ImpExpLog;
import com.wksc.framework.impexp.result.BaseResult;
import com.wksc.framework.impexp.task.Task;
import com.wksc.framework.impexp.util.ClassFactory;
import com.wksc.framework.impexp.util.SysParam;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;


/**
 * <p>
 * Title(文件名): TaskManage.java
 * </p>
 * <p>
 * Description(描述): 线程池的任务
 * </p>
 * <p>
 * Copyright(版权): Copyright (c) 2011
 * </p>
 * <p>
 * Company(公司): 成都四方信息技术有限公司
 * </p>
 * <p>
 * CreateTime(生成日期)：2011-10-20 下午03:02:25
 * </p>
 * 
 * @author wanglei
 * @version impExp_C13
 */
public class TaskManage {
	protected static final ImpExpLog LOG = ClassFactory
			.getLog(TaskManage.class);
	/** 等待队列的容量 */
	private static int waitingQueueSize = IEConstants.MAX_JOBS;
	/** 等待队列 */
	private static List<Task> waitingQueue = new LinkedList<Task>();
	/** 任务队列 */
	private static Map<String, ImpExpFuture<BaseResult>> taskList = new ConcurrentHashMap<String, ImpExpFuture<BaseResult>>(
			IEConstants.MAX_JOBS);
	/** 线程池 */
	private static ExecutorService threadPool = ImpExpThreadPool
			.newFixedThreadPool(IEConstants.THREAD_NUM);
	/** 暂停队列 */
	private static Map<String, Task> pauseList = new ConcurrentHashMap<String, Task>();
	/** 暂存已完成的任务名称的队列,用于将已完成任务从任务队列中移除 */
	private static List<String> finTaskList = new LinkedList<String>();
	/** 正在运行的任务类型Map,用于检测冲突 */
	private static Map<String, Boolean> runningTypeList = new ConcurrentHashMap<String, Boolean>();
	/** 导入类的互斥Map */
	private static Map<String, String> conflictGroupMap = null;

	private final static ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
	private final static Lock readLock = rwl.readLock();
	public final static Lock writeLock = rwl.writeLock();

	/**
	 * 方法描述：添加任务
	 * 
	 * @creator wanglei
	 * @param task
	 */
	public static synchronized int addTask(Task task) {

		//1. 如果队列满
		if (waitingQueueSize <= getWaitingNumInDb()) { //如果数据库中的等待数量已经超过设定值
			return IEConstants.TASK_LIST_FULL;
		}

		//2. 入队列前的验证,并且将task存入数据库
		if (null == task || !ClassFactory.getImpExpDao().addTask(task)) { //current_state保留0, job_state改为1
			return IEConstants.ADD_TASK_FAIL;
		}

		//将任务的状态设为"等待"
		task.setTaskStat(TaskStat.TASK_WAITING);
		//		task.setTaskStat4DbAndMem(TaskStat.TASK_WAITING);

		//尝试将任务添加至线程池运行
		addToRun(task);

		return IEConstants.ADD_TASK_SUCC;
	}

	/**
	 * 功能描述: checkeStyle
	 * 
	 * @Title: updateRefreshDate
	 * @Description: 更新ImportExport的刷新标记的值为数据库当前时间,表示当前这些任务由本机管理
	 * @creator Owner
	 * @return boolean true更新到数据 false更新数据时数据库异常 其他异常抛出
	 */
	public static boolean updateRefreshDate() {
		try {
			writeLock.lock();
			StringBuffer sb = new StringBuffer();
			for (Task task : waitingQueue) {
				sb.append(task.getBaseTemplate().getFileName() + ",");
			}

			Iterator<String> itTaskList = taskList.keySet().iterator();
			while (itTaskList.hasNext()) {
				String en = itTaskList.next();
				sb.append(en + ",");
			}
			//			Iterator<Entry<String, ImpExpFuture<BaseResult>>> it = taskList
			//					.entrySet().iterator();
			//			while (it.hasNext()) {
			//				Entry<String, ImpExpFuture<BaseResult>> en = it.next();
			//				sb.append(en.getKey() + ",");
			//			}

			Iterator<String> itPauseList = pauseList.keySet().iterator();
			while (itPauseList.hasNext()) {
				String en = itPauseList.next();
				sb.append(en + ",");
			}

			if (sb.length() > 0) {
				sb.deleteCharAt(sb.length() - 1);
				return ClassFactory.getImpExpDao().updateRefreshDate(
						sb.toString()) > -1; //更新到0条也可以
			}
			return true;
		} finally {
			writeLock.unlock();
		}
	}

	/**
	 * 方法描述：尝试将任务添加至线程池运行,若未能成功添加至线程池则添加至等待队列
	 * 
	 * @creator wanglei
	 * @param task
	 *            任务
	 */
	@SuppressWarnings("unchecked")
	public static synchronized void addToRun(Task task) {
		try {
			writeLock.lock();
			//正在运行的任务数未达到最大运行线程数并且没有同类型的其他任务正在运行
			//采用各个集群机器单独跑任务的形式,几个集群的机器,每个机器的运行中任务互不相关
			if (IEConstants.THREAD_NUM > taskList.size() // getRunningNum() 
					&& !checkTaskConflict(task)) {
				//检查数据库中数据是否已经被其他服务器占用,这个地方可能会导致一直取不到其他机器的任务
				if (!ClassFactory.getImpExpDao().updateTaskCurrentState(task,
						TaskStat.TASK_RUNNING)) { //current_state改为2,job_state保留1,返回true表示更新到记录 返回false表示未更新到记录
					return; //任务已经从waitingQueue中poll出来,这种任务不需要再加回到内存中
				}
				//提交至线程池并且放入任务列表中
				taskList.put(task.getBaseTemplate().getFileName(),
						(ImpExpFuture<BaseResult>) threadPool.submit(task));
				//设置future的taskName
				taskList.get(task.getBaseTemplate().getFileName()).setTaskName(
						task.getBaseTemplate().getFileName());
				//设置future的task对象
				taskList.get(task.getBaseTemplate().getFileName())
						.setTask(task);
			} else {
				//测试的时候发现:一条任务长时间断点,会出现自动任务复制.查一下原因
				if (!ClassFactory.getImpExpDao().updateTaskCurrentState(task,
						TaskStat.TASK_WAITING)) { //如果没有更新到任务,不在内存中保留
					return;
				}
				//将任务添至任务队列中
				((Queue<Task>) waitingQueue).offer(task);
			}
		} finally {
			writeLock.unlock();
		}
	}

	/**
	 * 方法描述：检查是否有同种任务正在运行
	 * 
	 * @creator wanglei
	 * @param task
	 *            任务
	 * @return
	 *         <p>
	 *         false-不冲突
	 *         </p>
	 *         <p>
	 *         true-冲突
	 *         </p>
	 */
	@SuppressWarnings("unchecked")
	private static boolean checkTaskConflict(Task task) {
		//hjj: 每一个任务的检查都需要查库,这个地方可以做缓冲
		/* 该任务存入冲突表的名称 */
		String conflictName = null;
		if (null == conflictGroupMap) {
			try {
				Method method = Class.forName(SysParam.getImpExpHeader())
						.getDeclaredMethod("getConflictGroupMap", null);
				conflictGroupMap = (Map<String, String>) method.invoke(null,
						new Object[0]);
			} catch (Exception e) {
				LOG.error(e.getMessage(), e);
			}
		}

		/* 确定该任务存入冲突表的名称 */
		if (conflictGroupMap.containsKey(task.getBaseTemplate().getTypeName())) {
			conflictName = conflictGroupMap.get(task.getBaseTemplate()
					.getTypeName());
		} else {
			conflictName = task.getBaseTemplate().getTypeName();
		}

		synchronized (runningTypeList) {
			if (runningTypeList.containsKey(conflictName)
					&& runningTypeList.get(conflictName)) {
				//理论上是不应该有冲突的,冲突最大的可能在于,内存放值后打个断点,再在数据库中将其删掉,
				//造成任务无法处理,这个时候就需要从数据库同步内存了
				LOG.info("Synchronizing running list begin. ");
				// hjj mod 121123,在检查冲突之前,首先从db同步runningTypeList
				List<?> lst = ClassFactory.getImpExpDao().getRunningList(
						task.getBaseTemplate().getFileName());
				if (lst == null) { // hjj 未从db中取得值
					throw new RuntimeException("获取数据库运行任务列表异常");
				}
				runningTypeList.clear();
				for (Object obj : lst) { //设置冲突
					ImportExport ie = (ImportExport) obj;
					runningTypeList.put(ie.getJobname(), true); //如:黑名单管理,如果有关联的话,这个地方需要把所有关联设置为true
				}
				LOG.info("Synchronizing running list successfully end. ");

				//				通过数据库判断是否冲突,从数据库同步完内存再比较
				if (runningTypeList.containsKey(conflictName)
						&& runningTypeList.get(conflictName)) {
					return true;
				} else {
					runningTypeList.put(conflictName, true);
					return false;
				}
			} else {
				runningTypeList.put(conflictName, true);
				return false; //不冲突
			}
		}

		//		if (ClassFactory.getImpExpDao().getQueueSize(TaskStat.TASK_WAITING,
		//				conflictName) == 0) {
		//			//			runningTypeList.put(conflictName, true);
		//			return false;
		//		} else {
		//			return true;
		//		}
	}

	/**
	 * 方法描述：清理任务队列中已完成的任务
	 * 该方法不保证能[完全]清理所有完成任务
	 * 
	 * @creator wanglei
	 */
	@SuppressWarnings("unused")
	private static void cleanTaskList() {
		String taskName = "";
		if (null == finTaskList || 0 == finTaskList.size()) {
			return;
		}
		//遍历完成列表清除已完成的任务
		for (int i = 0; i < finTaskList.size(); i++) {
			taskName = finTaskList.get(i);
			if (delTask(taskName)) {
				finTaskList.remove(taskName);
			}
		}
	}

	/**
	 * 方法描述：任务暂停的方法
	 * <p>
	 * 注意,该方法不接受空值, 调用此方法前先进行taskName的空验证
	 * </p>
	 * 
	 * @creator wanglei
	 * @param taskName
	 * @return
	 */
	public static boolean pauseTask(String taskName) {
		boolean isSucc = false;
		try {
			writeLock.lock();
			for (Task task : waitingQueue) {
				if (taskName.trim().equals(
						task.getBaseTemplate().getFileName().trim())
						&& waitingQueue.remove(task)) {
					pauseList.put(taskName, task);
					//					task.setTaskStat(TaskStat.TASK_PAUSED);
					task.setTaskStat4DbAndMem(TaskStat.TASK_PAUSED);
					isSucc = true;
					break;
				}
			}
			if (!isSucc) { //如果没有进内存才更新,如果进了内存,走普通流程
				//				ClassFactory.getImpExpDao().updateTaskState(taskName,
				//						TaskStat.TASK_PAUSED, true, true, false); //HJJ:是否应该更新为一个等待一个暂停
				updateDb(taskName, TaskStat.TASK_PAUSED.toInt(), ClassFactory
						.getI18n().getTaskpauseinfo(), false);
			}
			//			ClassFactory.getImpExpDao().updateTaskCurrentState(taskName,
			//					TaskStat.TASK_PAUSED);
			//			if (!isSucc) { //如果没有进内存
			//				ClassFactory.getImpExpDao().updateTaskJobState(taskName,
			//						TaskStat.TASK_PAUSED, true);
			//			}
		} finally {
			writeLock.unlock();
		}
		return isSucc;
	}

	/**
	 * 方法描述：任务取消的方法
	 * 
	 * @creator wanglei
	 * @param taskName
	 * @return
	 */
	public static boolean cancelTask(String taskName) {
		//		boolean isSucc = false;
		// 1.取消同时更新current_state和job_state
		// 如果a正在执行任务1, b触发取消,结果任务1显示为已经取消,但是实际上
		// a仍然还在后台执行中,该报错的报错,该完成的完成,只有在执行完这一批5k之后检查标识才会真正停止,但是也有可能再把标识置为完成
		// 2.取消只更新current_state
		// 如果a执行任务1时挂掉并且不启动,b触发取消,只能标识内存状态为取消,而不能标识任务状态为取消,界面上显示仍然是取消
		// 针对上面两个原因,增加通过判断refreshDate更新ImportExport.state
		try {
			writeLock.lock();
			//			ClassFactory.getImpExpDao().updateTaskState(taskName,
			//					TaskStat.TASK_CANCELLED, true, true, "");
			//			ClassFactory.getImpExpDao().updateTaskState(taskName,
			//					TaskStat.TASK_CANCELLING, true, true, false); //这个方法用下面这个方法来代替,下面这个方法是以前就有的
			updateDb(taskName, TaskStat.TASK_CANCELLING.toInt(), ClassFactory
					.getI18n().getTaskcancellinginfo(), false);

			//			ClassFactory.getImpExpDao().updateTaskCurrentState(taskName,
			//					TaskStat.TASK_CANCELLED);
			//			ClassFactory.getImpExpDao().updateTaskJobState(taskName,
			//					TaskStat.TASK_CANCELLED, true);
			if (null != taskList && null != taskList.get(taskName)) {
				//			isSucc = 
				taskList.get(taskName).cancel(true);
			}
			return true;
		} finally {
			writeLock.unlock();
		}
		//		return isSucc;
	}

	/**
	 * 方法描述：统一更新数据
	 * 
	 * @creator wanglei
	 * @param fileName
	 */
	private static void updateDb(String fileName, int taskState, String msg,
			boolean onlyUpdateSelf) {
		// 最后两个参数为-1表示现暂不需要该字段填入正确的值,采用-1占位
		ClassFactory.getImpExpDao().changeState(fileName, taskState, msg, -1,
				-1, true, onlyUpdateSelf);
	}

	/**
	 * 方法描述：任务恢复的方法
	 * 
	 * @creator wanglei
	 * @param taskName
	 * @return <p>
	 *         false - 恢复任务失败; true - 恢复任务成功
	 *         </p>
	 */
	public static boolean resumeTask(String taskName) {
		try {
			writeLock.lock();
			//			boolean isSucc = false;
			if (waitingQueueSize <= getWaitingNumInDb()) { //先判断队列是否已经满,如果队列已经满了不能讲pause的任务放到等待队列
				//这个地方应该使用数据库的等待数量,因为内存中的等待数量是每个机器小缓存中的数量
				return false;
			}
			//在数据库中将此任务改为等待,更新所有任务
			updateDb(taskName, TaskStat.TASK_WAITING.toInt(), ClassFactory
					.getI18n().getWaitinginfo(), false);
			//如果暂停列表为空或暂停列表中不包含传入的任务名则返回false
			if (null == pauseList || !pauseList.containsKey(taskName)) {
				//				ClassFactory.getImpExpDao().updateTaskState(taskName,
				//						TaskStat.TASK_PAUSED, true, true); 
				//				ClassFactory.getImpExpDao().updateTaskCurrentState(taskName,
				//						TaskStat.TASK_PAUSED);
				//				ClassFactory.getImpExpDao().updateTaskJobState(taskName,
				//						TaskStat.TASK_PAUSED, true);
				List<?> pausedTasks = ClassFactory.getImpExpDao()
						.findPausedTasks(false, true); //加载到本机
				//				TaskReload.reloadPausedTasks(pausedTasks);
				TaskReload.reloadRunningTasks(pausedTasks, false);
				//				return false;
				//			} else if (waitingQueueSize <= getWaitingNum()) {
				//				return false;
			}
			if (null == pauseList || !pauseList.containsKey(taskName)) {
				return false;
			}
			Task task = pauseList.get(taskName);
			task.setTaskStat(TaskStat.TASK_RESUME);
			addToRun(task);
			pauseList.remove(taskName);
			return true;
		} finally {
			writeLock.unlock();
		}
	}

	/**
	 * 方法描述：检查当前导入导出队列中的任务数
	 * 
	 * @creator wanglei
	 * @return
	 */
	public static int getWaitingNum() {
		readLock.lock();
		try {
			//			return ClassFactory.getImpExpDao().getQueueSize(null,
			//					TaskStat.TASK_WAITING);
			return waitingQueue.size();
		} finally {
			readLock.unlock();
		}
	}

	public static int getWaitingNumInDb() { //hjj mod 121123,集群环境下,判断导入文件冲突互斥问题
		readLock.lock();
		try {
			return ClassFactory.getImpExpDao().getQueueSize(null,
					TaskStat.TASK_WAITING);
		} finally {
			readLock.unlock();
		}
	}

	/**
	 * 方法描述：运行列表中的任务数量
	 * 
	 * @creator wanglei
	 * @return
	 */
	public static int getRunningNum() { //hjj mod 121123,集群环境下,判断导入文件冲突互斥问题
		//		readLock.lock();
		//		try {
		//		return ClassFactory.getImpExpDao().getQueueSize(null,
		//				TaskStat.TASK_RUNNING);
		//		} finally {
		//			readLock.unlock();
		//		}
		return taskList.size(); //每个机器的最大任务数之和=集群之后的可用任务数
	}

	/**
	 * 方法描述：从任务列表中删除指定任务
	 * 
	 * @param fileName
	 *            导入或导出的文件名,该值为数据库中主键,
	 *            在导入导出中此名称也看做为识别导入导出的任务名
	 * @creator wanglei
	 * @return
	 */
	public static boolean delTask(String fileName) {
		ClassFactory.getImpExpDao().syncCurrentState(fileName);
		if (null != taskList.remove(fileName)) {
			return true;
		}
		return false;
	}

	/**
	 * 方法描述：从任务列表中删除指定任务
	 * 
	 * @param fileName
	 *            导入或导出的文件名,该值为数据库中主键,
	 *            在导入导出中此名称也看做为识别导入导出的任务名
	 * @creator wanglei
	 * @return
	 */
	public static boolean delTask(String fileName, TaskStat stat) {
		//		ClassFactory.getImpExpDao().updateTaskCurrentState(fileName, stat);
		if (null != taskList.remove(fileName)) {
			return true;
		}
		return false;
	}

	/**
	 * 方法描述：此方法供当任务完成时的ImpExpFuture对象调用,
	 * 将已完成的任务名添加至完成队列中
	 * 
	 * @creator wanglei
	 * @param taskName
	 *            任务名称
	 * @return
	 */
	public static boolean addToFinList(String taskName) {
		boolean succ = false;
		try {
			succ = finTaskList.add(taskName);
		} catch (NullPointerException e) {
		}
		return succ;
	}

	/**
	 * 清空任务队列的方法
	 */
	public static void clearTasklist() {
		taskList.clear();
	}

	/**
	 * 方法描述：根据类型清理暂停队列的方法
	 * 
	 * @creator wanglei
	 * @param jobType
	 *            jobType
	 * @return 从内存暂停队列中删除的任务个数
	 */
	public static int cleanPauseList(String jobType) {
		int succNum = 0;
		try {
			writeLock.lock();
			int jobTypeInt = Integer.valueOf(jobType);
			Set<Entry<String, Task>> entrySet = pauseList.entrySet();
			for (Entry<String, Task> entry : entrySet) {
				if (jobTypeInt == entry.getValue().getBaseTemplate()
						.getTaskType().toTypeInt()) {
					pauseList.remove(entry.getKey());
					succNum += 1;
				}
			}
		} catch (Exception e) {
			LOG.error(e.getMessage(), e);
		} finally {
			writeLock.unlock();
		}
		return succNum;
	}

	@SuppressWarnings("unchecked")
	public static Queue<Task> getWaitingQueue() {
		return (Queue<Task>) waitingQueue;
	}

	public static synchronized void synchronizePauseList(List<?> pausedTasks) { //不允许两次进入
		try {
			if (pausedTasks == null || pausedTasks.isEmpty()) {
				return;
			}
			writeLock.lock();
			Iterator<String> it = pauseList.keySet().iterator();
			label_while: while (it.hasNext()) {
				String key = it.next();
				for (Object object : pausedTasks) {
					ImportExport imp = (ImportExport) object;
					if (imp.getFilename().equals(key)) {
						continue label_while;
					}
				}
				it.remove();
			}
		} catch (Exception e) {
			LOG.error("Sync pause list error:", e);
		} finally {
			writeLock.unlock();
		}
	}

	public static void clearConflictState(Task task) {
		String conflictName = "";
		synchronized (runningTypeList) {
			if (conflictGroupMap.containsKey(task.getBaseTemplate()
					.getTypeName())) {
				conflictName = conflictGroupMap.get(task.getBaseTemplate()
						.getTypeName());
			} else {
				conflictName = task.getBaseTemplate().getTypeName();
			}
			//hjj mod 121123,集群环境下,判断导入文件冲突互斥问题
			runningTypeList.put(conflictName, false);
			//			ClassFactory.getImpExpDao().setConflictState(conflictName, 0);
		}
	}

	/**
	 * 方法描述：用于重起服务器后重新将暂停的任务添加至暂停列表中
	 * 
	 * @creator wanglei
	 * @param task
	 */
	public static void addToPause(Task task) {
		try {
			writeLock.lock();
			pauseList.put(task.getBaseTemplate().getFileName(), task);
			task.setTaskStat4DbAndMem(TaskStat.TASK_PAUSED);
			//		task.setTaskStat(TaskStat.TASK_PAUSED);
		} finally {
			writeLock.unlock();
		}
	}

	/**
	 * 方法描述：根据传入的任务类型清理等待队列中该类型的所有任务
	 * 
	 * @creator wanglei
	 * @param jobType
	 *            导入还是导出任务
	 * @return 成功清理的任务数
	 */
	public synchronized static int cleanWaitingListByType(String jobType) {
		int succNum = 0;
		int jobTypeInt = -1;
		List<Task> toDeleteTasks = new ArrayList<Task>(getWaitingNum()); //这个地方以内存为准,避免查询数据库的消耗
		writeLock.lock();
		try {
			Iterator<Task> iter = waitingQueue.iterator();
			Task tmpTask = null;
			try {
				jobTypeInt = Integer.valueOf(jobType);
				while (iter.hasNext()) {
					tmpTask = iter.next();
					if (jobTypeInt == tmpTask.getBaseTemplate().getTaskType()
							.toTypeInt()) {
						toDeleteTasks.add(tmpTask); // 暂存于待删除列表
					}
				}

				for (Task task : toDeleteTasks) {
					if (waitingQueue.remove(task)) {
						succNum++;
					}
				}
			} catch (Exception e) {
				LOG.error(e.getMessage(), e);
			}
		} finally {
			writeLock.unlock();
		}
		return succNum;
	}
}
